package com.teamF.mockdata;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;

public class BridgeData {

    Properties props ;
    Producer<String, String> producer;

    public BridgeData(){
        props = new Properties();
        // Kafka服务端的主机名和端口号
        props.put("bootstrap.servers", "hadoop104:9092");
        // 等待所有副本节点的应答
        props.put("acks", "all");
        // 消息发送最大尝试次数
        props.put("retries", 0);
        // 一批消息处理大小
        props.put("batch.size", 16384);
        // 请求延时
        props.put("linger.ms", 1);
        // 发送缓存区内存大小
        props.put("buffer.memory", 33554432);
        // key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer(props);
    }

    public void generateDataAndSend(String paraId) throws InterruptedException {
        Random random = new Random();
        String paraTypeCode = UUID.randomUUID().toString();
        float paraValue = random.nextInt(70)+random.nextFloat();
        //每200毫秒一条数据
        for (int i = 0; i < 5; i++) {
            Date date = new Date();
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSSSSS");
            String timeString = format.format(date);
            String systemTime = timeString;
            String collectTime = timeString;
            int opt  = random.nextInt(2);
            switch (opt){
                case 0:
                    paraValue+=random.nextFloat();
                    break;
                case 1:
                    paraValue-=random.nextFloat();
                    break;
                default:
                    break;

            }
            String paraValueString = String.valueOf(paraValue);

            String data = systemTime+" "+collectTime+" "+paraId+" "+paraTypeCode+" "+paraValueString;
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), data));
            Thread.sleep(200);
        }

        producer.close();
    }





}
